@Greptile full review

Greptile Summary

This PR adds GPU-accelerated reading of Hadoop SequenceFiles by introducing a new GpuSequenceFileSerializeFromObjectExec that replaces SerializeFromObjectExec nodes backed by SequenceFile RDD scans at the physical plan level, and a supporting MultiFileCloudSequenceFilePartitionReader that reads files in parallel using a thread pool with an optional zero-copy GPU-concat combine mode. Performance benchmarks show ~2× speedup over the RDD baseline.

Key changes:

• New GpuSequenceFileSerializeFromObjectExecMeta detects SequenceFile RDD lineage via reflection and gates on a planning-time compression check, falling back to CPU for compressed inputs.
• New HostBinaryListBufferer and MultiFileCloudSequenceFilePartitionReader handle CPU-side decoding and H2D transfer of BytesWritable payloads into cuDF LIST<UINT8> device columns.
• Three new RapidsConf keys control reader type, parallelism cap, and enable/disable of the physical replacement.
• Integration and unit test coverage for basic reads, empty files, large batches, binary data, and missing/corrupt file handling.

Issues found:

• Data-loss risk: UnsupportedOperationException (thrown for compressed files at execution time) extends RuntimeException and is caught by the ignoreCorruptFiles handler, silently dropping compressed SequenceFiles when spark.sql.files.ignoreCorruptFiles=true. This is the most significant issue.
• Unreliable cycle detection: isSimpleSequenceFileRDD uses System.identityHashCode for the seen set — identity hash codes are not unique across JVM objects and can collide, incorrectly short-circuiting RDD lineage traversal and causing unnecessary CPU fallbacks.
• Double-close of SequenceFile.Reader: The closeOnExcept(reader) inner block and the outer try-finally both call reader.close(); the redundant close should be removed.
• Header version not checked in isCompressedSequenceFile: The isBlockCompressed boolean field is absent in SequenceFile format versions earlier than 5; reading it unconditionally can produce incorrect compression detection for old files.

Confidence Score: 2/5

• Not safe to merge as-is due to a data-loss risk when ignoreCorruptFiles=true encounters compressed SequenceFiles at execution time.
• The UnsupportedOperationException thrown for compressed files is silently swallowed as a "corrupted file" when ignoreCorruptFiles=true, dropping records without surfacing an error. While the planning-time check reduces the likelihood, it is not exhaustive (samples only one file per path), making data loss possible in production. The remaining issues (identity-hash cycle detection, double-close, old SequenceFile version header) are lower-severity but still warrant fixes before merge.
• sql-plugin/src/main/scala/com/nvidia/spark/rapids/sequencefile/GpuSequenceFileReaders.scala (ignoreCorruptFiles + double-close) and sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuSequenceFileSerializeFromObjectExecMeta.scala (identityHashCode cycle detection + SequenceFile header parsing).

Important Files Changed

Sequence Diagram

sequenceDiagram
    participant Driver
    participant Meta as GpuSequenceFileSerializeFromObjectExecMeta
    participant Exec as GpuSequenceFileSerializeFromObjectExec
    participant Factory as GpuSequenceFileMultiFilePartitionReaderFactory
    participant Reader as MultiFileCloudSequenceFilePartitionReader
    participant Hadoop as SequenceFile.Reader (CPU)
    participant GPU as cuDF / GPU

    Driver->>Meta: tagPlanForGpu()
    Meta->>Meta: isSimpleSequenceFileRDD() [reflection]
    Meta->>Meta: isCompressedSequenceFile() [header peek]
    Meta-->>Driver: willNotWorkOnGpu / OK

    Driver->>Exec: internalDoExecuteColumnar()
    Exec->>Factory: build reader factory
    Exec->>Reader: GpuDataSourceRDD per FilePartition

    loop For each PartitionedFile (threaded)
        Reader->>Hadoop: SequenceFile.Reader.nextRaw()
        Hadoop-->>Reader: raw key/value bytes
        Reader->>Reader: HostBinaryListBufferer.addBytes()
    end

    alt combineConf active (small files)
        Reader->>Reader: combineHMBs() — zero-copy chunk aggregation
    end

    Reader->>GPU: buildDeviceColumnFromChunks() → ColumnVector.concatenate()
    GPU-->>Exec: ColumnarBatch (LIST<UINT8> key + value)
    Exec-->>Driver: Iterator[ColumnarBatch]

_{Last reviewed commit: ed7fa84}

@greptileai full review

@greptileai full review

I am a little concerned about the general direction of this PR. If the RDDScan is a problem, then perhaps we should be trying to understand how to address that in a generic way instead of then sequence file reader code here. My main concern with the sequence file reader code is that there is a code change required for a user. The no code change is a fundamental principle of this project and I am concerned that we are violating that. This makes it much harder for a customer to switch over and also hard for a customer to fall back to the CPU if they need to.

I am a little concerned about the general direction of this PR. If the RDDScan is a problem, then perhaps we should be trying to understand how to address that in a generic way instead of then sequence file reader code here. My main concern with the sequence file reader code is that there is a code change required for a user. The no code change is a fundamental principle of this project and I am concerned that we are violating that. This makes it much harder for a customer to switch over and also hard for a customer to fall back to the CPU if they need to.

The customer is open to changing how they read the files. But I totally agree that we should keep the no-code-change principle here. It should be possible to introduce a rule that converts the RDDScan to a GPU file reader without changing the user code. I will work on it.

NOTE: release/26.02 has been created from main. Please retarget your PR to release/26.02 if it should be included in the release.

@greptileai full review

@greptileai full review